package com.wsjj.yjh;

import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.Map;

public class KeyedMapStateDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        env.enableCheckpointing(2000, CheckpointingMode.EXACTLY_ONCE);
        env.socketTextStream("1.94.41.70",7777)
                        .map(value -> {
                            String[] split = value.split(",");
                            return new WaterSensor(split[0],Long.valueOf(split[1]),Integer.valueOf(split[2]));
                        })
                                .assignTimestampsAndWatermarks(WatermarkStrategy.<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                        .withTimestampAssigner(new SerializableTimestampAssigner<WaterSensor>() {
                                            @Override
                                            public long extractTimestamp(WaterSensor element, long recordTimestamp) {
                                                return element.getTs();
                                            }
                                        }))
                                        .keyBy(value -> value.id)
                                                .process(new KeyedProcessFunction<String, WaterSensor, String>() {
                                                    MapState<Integer, Integer> mapState;

                                                    @Override
                                                    public void open(Configuration parameters) throws Exception {
                                                        mapState = getRuntimeContext().getMapState(new MapStateDescriptor<Integer, Integer>("mapState", Types.INT, Types.INT));
                                                    }

                                                    @Override
                                                    public void processElement(WaterSensor value, KeyedProcessFunction<String, WaterSensor, String>.Context ctx, Collector<String> out) throws Exception {
                                                        if(mapState.contains(value.getVc())){
                                                            mapState.put(value.getVc(),mapState.get(value.getVc())+1) ;
                                                        }else {
                                                            mapState.put(value.getVc(),1);
                                                        }
                                                        StringBuffer stringBuffer = new StringBuffer();
                                                        stringBuffer.append("======================================");
                                                        stringBuffer.append("\n");
                                                        stringBuffer.append("传感器id为" + value.getId() + "\n");
                                                        for (Map.Entry<Integer, Integer> entry : mapState.entries()) {
                                                            stringBuffer.append(entry.toString());
                                                            
                                                        }
                                                        out.collect(stringBuffer.toString());


                                                    }
                                                })
                                                        .print();


        env.execute();
    }
}
